昨天 D13 講為什麼要有可嵌入引擎，答案是 DataFusion 要成為能被 Spark、Ballista、新創的分析服務通用套用的執行引擎。可是講完架構動機還是虛的，一個引擎要能真的通用，得先解決一件很硬的事：這一台機器有 64 個 core，一句 SQL 要怎麼讓 64 個 core 都認真工作？
而且不能規劃時就分好、跑起來各憑本事，因為現實資料的分佈根本沒那麼乖，一定有人先跑完在旁邊納涼、有人被灌到爆卡到底，就跟工作總會有偷懶的
那今天要挖的就是這件事的答案，一篇 2014 年 SIGMOD 上的論文，Leis et al. 的 Morsel-Driven Parallelism: A NUMA-Aware Query Evaluation Framework for the Many-Core Age，HyPer 那篇，而 DataFusion、DuckDB、Velox 的並行執行血緣都追得到這裡。
先丟一個 D8 留下沒回答的謎題：
那天做 SortExec 實驗前，特別
SET datafusion.execution.target_partitions = 1
為什麼？
因為 DataFusion 天生就是多核並行執行，如果不固定的話，量到的每個時間都會被多個 worker 各自跑完再加總平均掉。這個多核並行 DataFusion 不是始祖，而是這篇論文奠的框架，今天就來理解這框架吧
四個問題：
2014 年之前，OLAP 引擎要並行執行主要有兩派做法，其實骨架都是同一套。
Volcano exchange operator（Graefe 1990）
做法是在 plan tree 裡插一個 Exchange 節點，這個節點負責把資料流拆成 N 份，分給 N 個 worker thread 各跑一份 sub-plan
Plan-driven parallelism
編譯期或規劃期就決定要開幾個 thread，每個 thread 吃哪塊資料，跑起來就照著計劃走
兩者的共通點是分工這件事在 query 開始前就決定好
想像一個場景：主管在早上開會時把今天的 100 份訂單分給 10 個同事，每人 10 份，然後就散會，中間不再協調，這在訂單都差不多重的時候還好，可是現實工作不長這樣：
結果就是下午 3 點有 7 個人做完在納涼，剩 3 個人加班到晚上，整個團隊被最慢那個人拖到底。
這就是 Volcano exchange 的痛點：靜態切分無法適應執行時的不均勻
還有一個 2014 年才變得很痛的問題，就是 NUMA
NUMA（Non-Uniform Memory Access）是說多 socket 的伺服器上，每顆 CPU 都有自己這邊的記憶體。
CPU 存取自己 socket 上的記憶體很快，跨到另一個 socket 就慢（30–50% 的 latency penalty）。Volcano exchange 完全不管這件事，它把 worker 開起來就丟到某個 core 上跑，要處理的資料可能剛好在對面 socket 的記憶體，每次讀取都要跨界，會很費效率。
論文的核心翻轉是：分工不要在計劃時決定，搬到執行時決定
具體怎麼做？
把資料切成小塊，論文叫 morsel（一小口的意思），大約 100K rows 一塊
中央有個 dispatcher，手上握著這個 query 所有的 morsel 清單，每個 worker thread 做完手上的活就跟 dispatcher 說「我沒事做了，給我下一塊」，dispatcher 派一個 morsel 過去。
換個生活比喻，Volcano 是主管開會時分工，morsel-driven 是壽司帶
師傅（worker）看到帶子上有壽司（morsel）就伸手拿一份，拿完做完再等下一份，永遠不會有「這個師傅閒著、那個師傅忙翻」的狀況，因為忙完的自然會去拿下一份。
再加一個關鍵條件，dispatcher 派 morsel 的時候優先派本 NUMA node 上的 morsel
Worker A 住在 socket 0，那 socket 0 記憶體上的 morsel 都優先派給他，socket 1 的 morsel 派給 socket 1 上的 worker。只有當 socket 0 的活都做完了、A 還想工作，才去 socket 1 偷一塊來做，這個動作叫 work-stealing。
論文的骨架其實就四個字。
morsel 要多小？
(Source: Leis et al., SIGMOD 2014, Figure 2 或 Figure adapted from Leis et al.(2014))
一句 SQL 的物理計劃通常長這樣：Scan → Filter → HashJoinProbe → HashAggBuild → HashAggFinal
其中像是 hash join 的 build 階段、aggregation 這種必須等所有資料到齊才能往下游送的運算，論文稱為 pipeline breaker。
從 scan 開始，一路流到遇到 breaker 為止，中間的算子彼此之間不物化中間結果、資料像流水線一樣一路流過去（HyPer 用 JIT 編譯讓這段甚至比向量化引擎更進一步不物化），這一段就是一個 pipeline。而 breaker 本身的工作，就是把這段 pipeline 流出來的資料先物化暫存，等全部到齊之後，才讓下一段 pipeline 開始跑。
Pipeline 內是流式，pipeline 之間才物化，這是 morsel-driven 排程的最小單元，dispatcher 就是在 pipeline 這個尺度上派 morsel。
Dispatcher 手上有：這個 query 所有 pipeline、每個 pipeline 剩下多少 morsel、每個 morsel 屬於哪個 NUMA node、每個 worker 現在住在哪個 socket。
派工原則兩條：
因為分工搬到執行時決定，一個 query 跑到一半、另一個急件進來時，可以馬上把幾個 worker 抽走支援急件，跑完再回來，不用把整個 query 打掉重新規劃。這對真正的 multi-tenant 場景很重要（例如共用一台 db 給很多使用者）。
上一節提到 HashJoinProbe，這裡來解釋一下，因為 morsel-driven 論文 §4 的核心例子就是 hash join，之後講 join 演算法也會反覆用到。
Probe（探測）就是 hash join 執行時「查表」的那個動作。
Hash join 天然分兩個角色：
論文用 R ⋈A S ⋈B T 當例子（R join S on 欄位 A，S join T on 欄位 B），實際執行順序是：
1. 先掃描 T，建出 HT(T) ← build
2. 再掃描 S，建出 HT(S) ← build
3. 最後掃描 R，拿 R 的每一筆資料
去 HT(S) 裡查（probe）、
查完再去 HT(T) 裡查（probe） ← probe pipeline
所以第三條 pipeline 才叫 probe pipeline，它得等前兩個 hash table 都建好之後才能開始，把 R 一筆一筆拿去探測兩張已經建好的表，找出真正符合 join 條件的組合。
順便說 algebraic plan 是什麼意思
join、filter、projection 這些運算，數學上遵守關聯代數（relational algebra）的規則（像加法乘法一樣可以交換、結合），這讓 optimizer 可以合法調整運算順序（例如決定先 join 哪兩張表）來找最快的執行方式，所以論文裡看到的那棵樹叫 algebraic plan，是邏輯上要做什麼的表達，還沒決定怎麼平行跑
這棵樹跟 morsel-driven 執行的橋樑就在這裡
論文接下來做的事，是把樹上每一段 pipeline segment（一段不被 hash table 這種 pipeline breaker 打斷的路徑）分配給多個 thread 去平行跑，這邊對應論文 Figure 2 右邊那張同一棵樹被畫成多條顏色平行線的圖。
三個 build pipeline 的 morsel 可以各跑各的、互相不知道對方存在，probe pipeline 得等它們都完成才能開始。
Hash table 之所以是 pipeline breaker，就是因為它是 build pipeline 跟 probe pipeline 之間天然的同步點，這也是 dispatcher 派 morsel 時要一個 pipeline 派完再派下一個的原因。
論文在2014年揪發表了，十年後 DataFusion 把它搬進來
原本的做法：DataFusion 早期跟 Volcano 一樣，掃 Parquet 時在規劃階段就把檔案固定分配給每個 worker，10 個檔案配 10 個 core，一人一份，各做各的到底。
TPC-H 上完全看不出問題，因為 TPC-H 的資料分佈很均勻，論文自己在 §6 也提到這個侷限，可是換成 ClickBench（真實查詢 log 改編）就現形：
結果就是某個 thread 分到特別重的活，其他人做完在旁邊空轉，整體被最慢那個拖死，跟 §1 那個主管早上分完工的比喻一模一樣。
解法就是原樣把 morsel-driven 搬進來：
RepartitionExec、CoalescePartitionsExec
這件事是一連串 PR 慢慢推進的，是個大重構拆小步走的示範：
#20529 (提議：借用 HyPer 論文的 morsel-driven 概念)
↓
#20820 (先設計 Morselizer API)
↓
#21327 (實作 Morsel API，改寫 Parquet reader)
↓
#21342 (改寫 FileStream 使用 Morsel API)
↓
#21351 (真正加上動態 work-stealing，最終合併)
實測效果：ClickBench 上 33 個查詢變快，部分查詢加速達 2.2 到 2.3 倍，平均查詢時間從 634 ms 降到 581 ms，skew 越大的查詢 morsel-driven 收益越明顯。
Work-stealing 有一個很少被明說的隱含假設：所有 worker 都在同一個 process 裡，可以共享一個記憶體內的 queue，但這個假設在單機上成立，也是為什麼論文從頭到尾不用討論。
可是 DataFusion 不只被單機用，它也被拿去做分散式執行引擎的核心（Ballista、datafusion-distributed），每個 partition 跑在不同機器 / 不同 process。這時候：
結果是資料量直接膨脹成 partition 數倍，同一份檔被讀好幾次，這已經不是效能問題，是正確性錯誤（correctness bug）。
最後的解法是加一個 session-level 開關 enable_file_stream_work_stealing，讓分散式執行器可以把這個單機專屬優化明確關掉。
這裡有個抽象一層問題值得記下來：一個 locality-aware + elastic 的排程機制，優點建立在 全域狀態可見、可共享記憶體 這個隱含假設上，但當場景從「單機」搬到「跨 process / 跨 cluster / 跨 cloud」，這個假設會被打破，而且往往不是效能打折，是正確性直接失效，這種時候不能自動 fallback，得有明確的 opt-out 開關。
論文讀完，最自然的下一個問題是：DataFusion 是不是就長這樣？
答案不是「是」或「不是」，得把 morsel-driven 拆成三個零件分開看。
有，而且 Parquet 這裡切得很細
DataFusion 會把單一個大 Parquet 檔切開分給多核。開關叫 repartition_file_scans，門檻是 repartition_file_min_size，預設總量超過 10MB 才切，避免對一堆小檔做無謂的切分。實作方式是同一個檔案用不同的 byte range 出現在多個 partition 裡。
切的精度也很有意思：row group 屬於哪一段，是看它的中點落在哪個 range 裡，所以 range 不需要對齊 row group 邊界。一個 2GB 的檔案可以直接無腦砍成四段 512MB，只要裡面有幾個大小相近的 row group 就行，好處是不用先讀 metadata 就能決定怎麼切。這個做法是從 Spark 的 splitFiles 學來的。
所以「把大工作拆成小塊」這件事，DataFusion 做得很徹底。
有，但偷的東西不一樣
work-stealing 也有，來自 tokio 的 scheduler。但它偷的單位是 async task，不是 morsel。翻成白話：偷的是「某條執行流的下一次推進」，不是「一塊還躺在那裡沒人處理的資料」。
這個沒有
這才是論文的招牌，也是 DataFusion 唯一缺的那塊。
DataFusion 的切法全部在 planning 階段就定案。planner 看 target_partitions 和檔案大小，算出要切幾份、每份哪個 range，然後執行期間再也不變。跑到一半沒有任何機制可以說「第 3 條落後了，把它剩下的工作分一點出去」。
morsel-driven 的核心恰恰是這個：dispatcher 手上握著一堆還沒發出去的 morsel，誰做完誰來領，而且領的時候會考慮 NUMA 距離。
叫 Exchange operator，實作是 RepartitionExec。planner 決定目標 partition 數，RepartitionExec 把資料重新分配成那麼多條輸出 partition，每條就是一條獨立的執行流。
這個設計源自 1989 年 Volcano 那篇談平行化的論文，比今天讀的這篇早了二十幾年。商業系統幾乎都是這一派，DuckDB 是少數例外。
但代價在哪呢？
靜態切分最怕的是 skew
切成 16 份，如果其中一份的 filter 選擇率特別低、實際要處理的資料量大很多，那個 core 會拖到最後，剩下 15 個 core 只能空等。執行期沒有再平衡的手段。
這正是 morsel-driven 想解的問題。所以這不是「論文講的東西沒用」，而是論文解的那個問題，DataFusion 選擇用工程上更簡單的方式繞過去，並且接受 skew 的風險。benchmark 上兩者拉不開，所以這個選擇到目前為止是划算的。
那 exchange 這條路在執行層具體怎麼把資料推進下去？這是 D16 的事。
現在可以回答 D8 那句 target_partitions = 1 為什麼是實驗設備裡最關鍵的一行了。
DataFusion 預設會照 CPU 核數開多個 partition，每個 partition 對應一個 worker thread，然後 morsel-driven 派工在這幾個 worker 之間動態切。你量 elapsed_compute 的時候，這個數字是多個 worker 各自 CPU 時間的加總，不是 wall clock。
如果不把 partition 打成 1，看到 SortExec 花了 10 秒，其實可能是 8 個 worker 各花 1.25 秒的加總，或者 4 個 worker 各花 2.5 秒 + 4 個閒著，中間到底是量到排序演算法本身的時間還是 morsel 派工加 work-stealing 的時間，實際上根本分不出來。
把 target_partitions = 1 之後，plan 裡只剩一條 pipeline、一個 worker、一份 morsel 從頭吃到尾，這時候量到的 elapsed_compute 就是這個演算法在單核上要花多少時間的數字
今天讀了一篇 2014 年的論文，看它在 2024 年被 DataFusion 拿去用、又在分散式場景意外炸出一個 bug。這篇之所以重要，不是因為它提出了什麼新演算法，而是它建立了一個看待多核並行執行的框架：把分工搬到執行時、以 morsel 為單位、以 NUMA locality 為優先、以 work-stealing 為 fallback。
留一個沒有標準答案的問題：morsel 100K rows 是 2014 年 Nehalem-EX（L3 cache 24 MB）上得出的結論，2026 年 Zen 5 / Apple M4（L3 cache 32 到 64 MB）上還適用嗎？雲上的單 socket 大機器（AWS Graviton c7g.16xlarge）呢？ 論文證明的是存在一個峰值，但這個峰在 100K是量測結果，會隨 cache 大小、記憶體頻寬、每列位元組數漂動。如果真的要設計一個，應該要怎麼設計一個實驗把新硬體上的峰位量出來？提示，morsel 太小 dispatcher 開銷會爆，太大負載均衡失效，兩邊都要能看到才叫量到一個峰。
那就明天見～
iThome鐵人賽
看影片追技術